package com.swak.hazelcast;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import com.swak.Constants;
import com.swak.codec.Encodes;
import com.swak.registry.ChildListener;
import com.swak.registry.FailbackRegistry;
import com.swak.registry.NotifyListener;
import com.swak.registry.URL;
import com.swak.utils.CollectionUtils;
import com.swak.utils.Lists;
import com.swak.utils.Sets;

public class HazelcastRegistry extends FailbackRegistry {

	private final ConcurrentMap<String, Set<URL>> urlsRegistered = new ConcurrentHashMap<>();
	private final ConcurrentMap<String, Set<ChildListener>> listeners = new ConcurrentHashMap<>();
	private final ConcurrentMap<URL, Set<String>> anyModules = new ConcurrentHashMap<>();
	private final ConcurrentMap<URL, Set<String>> anyServices = new ConcurrentHashMap<>();
	private final ConcurrentMap<URL, ConcurrentMap<NotifyListener, ChildListener>> zkListeners = new ConcurrentHashMap<>();

	public HazelcastRegistry(URL url) {
		super(url);
	}

	private void doNotify(String path) {
		Set<ChildListener> _listeners = listeners.get(path);
		if (_listeners == null || _listeners.size() == 0) {
			return;
		}

		List<String> childs = getChildrenPaths(path);
		for (ChildListener listener : _listeners) {
			listener.childChanged(path, childs);
		}
	}

	@Override
	public void subscribe(URL url, NotifyListener listener) {
		super.subscribe(url, listener);
		this.doSubscribe(url, listener);
	}

	@Override
	public void doRegister(URL url) {
		for (String path : url.toCategoryPaths()) {
			Set<URL> urls = urlsRegistered.computeIfAbsent(path, (key) -> {
				return Sets.newConcurrentSet();
			});
			urls.add(url);
			doNotify(path);
		}
	}

	@Override
	public void doUnregister(URL url) {
		for (String path : url.toCategoryPaths()) {
			Set<URL> urls = urlsRegistered.computeIfAbsent(path, (key) -> {
				return Sets.newConcurrentSet();
			});
			urls.remove(url);
			doNotify(path);
		}
	}

	@Override
	public void doSubscribe(URL url, NotifyListener listener) {
		if (Constants.Subscribe_Any_Value.equals(url.getModule())) {
			this.doAnyModuleSubscribe(url, listener);
		} else if (Constants.Subscribe_Any_Value.equals(url.getService())) {
			this.doAnyServiceSubscribe(url, listener);
		} else {
			this.doCategorySubscribe(url, listener);
		}
	}

	@Override
	public void doUnsubscribe(URL url, NotifyListener listener) {
	}

	private void doAnyModuleSubscribe(URL url, NotifyListener listener) {
		ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url,
				k -> new ConcurrentHashMap<>());
		ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (parentPath, currentChilds) -> {
			for (String child : currentChilds) {
				child = Encodes.urlDecode(child);
				Set<String> services = anyModules.get(url);
				if (services == null) {
					services = anyModules.computeIfAbsent(url, (s) -> {
						return Sets.newConcurrentSet();
					});
				}
				if (!services.contains(child)) {
					services.add(child);
					subscribe(URL.valueOf(url).addParameter(Constants.Module_Key, child), k);
				}
			}
		});
		List<String> currentChilds = addChildListener(url.toRootPath(), zkListener);
		if (CollectionUtils.isNotEmpty(currentChilds)) {
			for (String child : currentChilds) {
				child = Encodes.urlDecode(child);
				Set<String> services = anyModules.get(url);
				if (services == null) {
					services = anyModules.computeIfAbsent(url, (s) -> {
						return Sets.newConcurrentSet();
					});
				}
				if (!services.contains(child)) {
					services.add(child);
					subscribe(URL.valueOf(url).addParameter(Constants.Module_Key, child), listener);
				}
			}
		}
	}

	private void doAnyServiceSubscribe(URL url, NotifyListener listener) {
		ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url,
				k -> new ConcurrentHashMap<>());
		ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (parentPath, currentChilds) -> {
			for (String child : currentChilds) {
				child = Encodes.urlDecode(child);
				Set<String> services = anyServices.get(url);
				if (services == null) {
					services = anyServices.computeIfAbsent(url, (s) -> {
						return Sets.newConcurrentSet();
					});
				}
				if (!services.contains(child)) {
					services.add(child);
					subscribe(URL.valueOf(url).addParameter(Constants.Service_Key, child), k);
				}
			}
		});
		List<String> currentChilds = addChildListener(url.toRootPath(), zkListener);
		if (CollectionUtils.isNotEmpty(currentChilds)) {
			for (String child : currentChilds) {
				child = Encodes.urlDecode(child);
				Set<String> services = anyServices.get(url);
				if (services == null) {
					services = anyServices.computeIfAbsent(url, (s) -> {
						return Sets.newConcurrentSet();
					});
				}
				if (!services.contains(child)) {
					services.add(child);
					subscribe(URL.valueOf(url).addParameter(Constants.Service_Key, child), listener);
				}
			}
		}
	}

	private void doCategorySubscribe(URL url, NotifyListener listener) {
		List<URL> urls = new ArrayList<>();
		for (String path : url.toCategoryPaths()) {
			ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url,
					k -> new ConcurrentHashMap<>());
			ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (parentPath,
					currentChilds) -> HazelcastRegistry.this.notify(url, k, toUrls(url, currentChilds)));
			List<String> children = addChildListener(path, zkListener);
			if (children != null) {
				urls.addAll(toUrls(url, children));
			}
		}
		this.notify(url, listener, urls);
	}

	private List<String> addChildListener(String path, ChildListener zkListener) {
		Set<ChildListener> childs = listeners.computeIfAbsent(path, (s) -> {
			return Sets.newConcurrentSet();
		});
		childs.add(zkListener);
		return getChildrenPaths(path);
	}

	private List<String> getChildrenPaths(String path) {
		Set<URL> registereds = this.urlsRegistered.get(path);
		if (registereds == null || registereds.size() == 0) {
			return null;
		}
		List<String> providers = Lists.newArrayList();
		for (URL url : registereds) {
			providers.add(url.toFullString());
		}
		return providers;
	}

	private List<URL> toUrls(URL consumer, List<String> providers) {
		List<URL> urls = new ArrayList<>();
		if (CollectionUtils.isNotEmpty(providers)) {
			for (String provider : providers) {
				URL url = URL.valueOf(Encodes.urlDecode(provider));
				if (consumer.match(url)) {
					urls.add(url);
				}
			}
		}
		return urls;
	}
}
